#include "transContentPortStream.hpp"

////////////////////////////////////////////////////////////
/// initinatize the stream record
///		send sync protocol to remote, use block style
///
/// parameters:
///		type: called by accept create stream port(1) or connect stream port(2)
///
/// return:
///		1: sucessfully
///		0: failed
/////////////////////////////////////////////////////////////
int TransContentPortStream::inti_sync(int type){
	char buffer[44] = {0};
	timespec ts;
	debug_enter();

	/// construct the sync patket
	logd_info("create the steam port sync packet");
	*(short*(buffer)) = (short)35; //total len
	buffer[2] = (char)message_type_accept-connect; //main code
	*(NodeAddr*(buffer+3)) = target; //target
	buffer[11] = CLIENT_STACK_FLAG_BITS_FIXEDPATH; //flag
	*(short*(buffer+12)) = (short)23; //len
	buffer[14] = (char)message_type_accept-connect; //main code
	buffer[15] = (char)message_type_accept-connect_sub_streamStart; //sub code
	*(int*(buffer+16)) = TransContentAssis::INTL2N(remotePort); //remotePort
	*(int*(buffer+20)) = TransContentAssis::INTL2N(localPort); //localPort
	*(NodeAddr*(buffer+24)) = TransContentSetting::getLocalAddrNetBytes(); // local addr
	*(int*(buffer+32)) = TransContentAssis::INTL2N(streamOut->getStreamStartValue()); //stream id
	buffer[36] = (char)streamIn->getStreamBufferSize(); //buffer size

	/// send to stack
	logd_info("construct packet finished, start send packet to stack");
	TransContentDispatcher::getInstance()->sendDataDataToStack(buffer, 37);

	if(type == 2){
		logd_info("stream sync method use call by tpye==2, so no need to wait response");
		return 1;
	}

	// wait
	ts = TransContentAssis::TimeTp2Ts(timeout);
	pthread_mutex_lock(&waitMutex);
	if(pthread_cond_timedwait(&waitCond, &waitMutex, &ts) == ETIMEDOUT){
		logd_error("wait remote steam start timeout!");
		return 0;
	}
	pthread_mutex_unlock(&waitMutex);

	logd_info("stream port sync sucessfully");
	return 1;
}

int TransContentPortStream::response(char* content, int index){
	int mainCode = content[0];
	int subCode = content[1];
	if(mainCode != message_type_out_data_stream || mainCode != message_type_accept-connect){
		logd_error("not support the mainCode, the content should be post to next port, message code: %d", mainCode);
		return RESPONSE_RESULT_POST_CONITNUE;
	}

	switch(subCode){
		case message_type_accept-connect_sub_streamStart:
			logd_info("sub message: stream port init");
			streamStartResponse(content);
			break;
		case message_type_data_stream_sub_portSteate: //port exist
			logd_info("sub message: port heart beat");
			heartBeatResponse(content);
			break;
		case message_type_data_stream_sub_portClose: //port close
			logd_info("sub message: port closed");
			portCloseResponse(content);
			break;
		case message_type_data_stream_sub_dataIn: //in data
			logd_info("sub message: data in");
			dataInResponse(content);
			break;
		case message_type_data_stream_sub_dataACK: //remote data ack
			logd_info("sub message: data ack");
			dataAckResponse(content);
			break;
		default:
			logd_info("not support sub message for port stream");
	}

	logd_info("start to release the response buffer");
	TransContentMemory::getInstance()->releaseBuffer(index);
	return RESPONSE_RESULT_NOACTION;
}

///////////////////////////////////////////////////////////
/// check port or target
///		then set the streamSize and streamId and signal wait thread
///
/// paramters:
///		conent: the response buffer
///////////////////////////////////////////////////////////
void TransContentPortStream::streamStartResponse(char* content){
	debug_enter();
	int _remotePort = TransContentAssis::INTN2L(*(int*(content+6))); //remote port
	NodeAddr _remoteAddr = TransContentAssis::NodeAddrN2L(*(NodeAddr*(content+10))); //remote addr
	int _stramStartValue = TransContentAssis::INTN2L(*(int*(content+18))); //stream start value
	int _steamBufferSize = content[22]; //buffer size
	logd_info("local value: remotePort:%d, remoteAddr:%lld, content value: remotePort:%d, remoteAddr:%lld", remotePort, target.serial_id, _remotePort, _remoteAddr.serial_id);
	streamIn->setRemoteStreamStart(_stramStartValue);
	streamOut->setlocalBufferSize(_steamBufferSize);
	logd_info("stream start response set stream start value and buffersize finished");
	debug_leave();
}

void TransContentPortStream::dataInResponse(char* content){
	logd_info("stream port data in protocol passed to stream in memroy");
	streamIn->response(content);
}

void TransContentPortStream::dataAckResponse(char* content){
	logd_info("stream port data ack protocol passed to stream out memory");
	steamOut->response(content);
}

void TransContentPortStream::portCloseResponse(char* content){
	logd_info("stream port received close port protocol, set canBeRemoved to 1");
	canBeRemoved = 1;
}

void TransContentPortStream::heartBeatResponse(char* content){
	char buffer[36] = {0};
	logd_info("steam port received heart beat protocol, construct heart beat rsponse packet, and send back");

	/// construct the packet
	*(short*(buffer)) = (short)31;
	buffer[2] = (char)message_type_out_data_stream; //mainCode
	*(NodeAddr*(buffer+3)) = TransContentAssis::NodeAddrL2N(target); //target
	buffer[11] = CLIENT_STACK_FLAG_BITS_FIXEDPATH; //fixed flag
	*(short*(buffer+12)) = (short)19;
	buffer[14] = (char)message_type_out_data_stream;; //main code
	buffer[15] = (char)message_type_data_stream_sub_portSteate_response; //sub code
	*(int*(buffer+16)) = TransContentAssis::INTL2N(remotePort); //remote port
	*(int*(buffer+20)) = TransContentAssis::INTL2N(localPort); //local port
	*(NodeAddr*(buffer+24)) = TransContentSetting::getLocalAddrNetBytes(); //local addr
	buffer[32] = (char)1; //exist

	///send to stack
	TransContentDispatcher::getInstance()->sendDataDataToStack(buffer, 32);
	logd_info("stream heart beat response finished");
	return;
}
